package cn.edu.spark.sql

import org.apache.spark.sql.SparkSession

import java.lang.Thread.sleep
import java.util.Properties

object SQLDataSourceExample {
  case class Person(name: String, age: Long)


  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("Spark SQL data sources example")
      .master("local[*]")
      .getOrCreate()

    //    runBasicDataSourceExample(spark)
    //    runGenericFileSourceExample(spark)
    //    runParquetExample(spark)
    //    runJsonDatasetExample(spark)
    //    runCsvDatasetExample(spark)
    //    runTextDatasetExample(spark)
    runJdbcDatasetExample(spark)

    sleep(300000)
    spark.stop()
  }

  def runBasicDataSourceExample(spark: SparkSession): Unit = {
    // generic_load_save_functions
    val usersDF = spark.read.load("spark-sql-examples/src/main/resources/users.parquet")
    usersDF.show()
    usersDF.select("name", "favorite_color").write.save("tem/namesAndFavColors.parquet")

    // manual_load_options
    val peopleDF = spark.read.format("json").load("spark-sql-examples/src/main/resources/people.json")
    peopleDF.show()
    peopleDF.select("name", "age").write.format("parquet").save("tem/namesAndAges.parquet")


    // direct_sql
    val sqlDF = spark.sql("SELECT * FROM parquet.`spark-sql-examples/src/main/resources/users.parquet`")
    sqlDF.show()

    // write_sorting_and_bucketing
    usersDF.write.bucketBy(42, "name").sortBy("name").saveAsTable("users_bucketed")
    // saveAsTable 保存到数仓中

    // write_partitioning
    usersDF.write.partitionBy("favorite_color").saveAsTable("users_partitioned")

    // write_partition_and_bucket
    usersDF
      .write
      .partitionBy("favorite_color")
      .bucketBy(42, "name")
      .saveAsTable("users_partitioned_bucketed")

    // 删除仓库中的表
    spark.sql("DROP TABLE IF EXISTS users_bucketed")
    spark.sql("DROP TABLE IF EXISTS users_partitioned")
    spark.sql("DROP TABLE IF EXISTS users_partitioned_bucketed")
  }

  def runGenericFileSourceExample(spark: SparkSession): Unit = {
    // enable ignore corrupt files
    spark.sql("set spark.sql.files.ignoreCorruptFiles=true")
    // dir1/file3.json is corrupt from parquet's view
    val testCorruptDF = spark.read.parquet("spark-sql-examples/src/main/resources/dir1/",
      "spark-sql-examples/src/main/resources/dir1/dir2/") // format("parquet").load(paths: _*)
    testCorruptDF.show()
    // +-------------+
    // |         file|
    // +-------------+
    // |file1.parquet|
    // |file2.parquet|
    // +-------------+
  }

  def runParquetExample(spark: SparkSession): Unit = {
    // basic_parquet_example
    val peopleDF = spark.read.json("spark-sql-examples/src/main/resources/people.json")
    // DataFrames can be saved as Parquet files, maintaining the schema information
    peopleDF.write.parquet("tem/people.parquet")

    // Parquet files are self-describing so the schema is preserved
    val parquetFileDF = spark.read.parquet("tem/people.parquet")

    // Parquet files can also be used to create a temporary view and then used in SQL statements
    parquetFileDF.createOrReplaceTempView("parquetFile")
    val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")

    import spark.implicits._
    namesDF.map(attributes => "Name: " + attributes(0)).show()
    // +------------+
    // |       value|
    // +------------+
    // |Name: Justin|
    // +------------+

    // schema_merging
    // Create a simple DataFrame, store into a partition directory
    val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
    squaresDF.write.parquet("tem/data/test_table/key=1")

    // Create another DataFrame in a new partition directory
    val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
    cubesDF.write.parquet("tem/data/test_table/key=2")

    // Read the partitioned table
    val mergedDF = spark.read
      .option("mergeSchema", "true")
      .parquet("tem/data/test_table")
    mergedDF.printSchema()
    // root
    //  |-- value: int (nullable = true)
    //  |-- square: int (nullable = true)
    //  |-- cube: int (nullable = true)
    //  |-- key: int (nullable = true)
    mergedDF.show()
    //  +-----+------+----+---+
    //  |value|square|cube|key|
    //  +-----+------+----+---+
    //  |    1|     1|null|  1|
    //  |    2|     4|null|  1|
    //  |    3|     9|null|  1|
    //  |    4|    16|null|  1|
    //  |    5|    25|null|  1|
    //  |    6|  null| 216|  2|
    //  |    7|  null| 343|  2|
    //  |    8|  null| 512|  2|
    //  |    9|  null| 729|  2|
    //  |   10|  null|1000|  2|
    //  +-----+------+----+---+
  }

  def runJsonDatasetExample(spark: SparkSession): Unit = {
    // json_dataset
    import spark.implicits._

    // A JSON dataset is pointed to by path.
    // The path can be either a single text file or a directory storing text files
    val path = "spark-sql-examples/src/main/resources/people.json"
    val peopleDF = spark.read.json(path)

    // The inferred schema can be visualized using the printSchema() method
    peopleDF.printSchema()
    // root
    //  |-- age: long (nullable = true)
    //  |-- name: string (nullable = true)

    // Creates a temporary view using the DataFrame
    peopleDF.createOrReplaceTempView("people")

    // SQL statements can be run by using the sql methods provided by spark
    val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
    teenagerNamesDF.show()
    // +------+
    // |  name|
    // +------+
    // |Justin|
    // +------+

    // Alternatively, a DataFrame can be created for a JSON dataset represented by
    // a Dataset[String] storing one JSON object per string
    val otherPeopleDataset = spark.createDataset(
      """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" ::
        """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" ::
        Nil)
    val otherPeople = spark.read.json(otherPeopleDataset)
    otherPeople.show()
    // +---------------+----+
    // |        address|name|
    // +---------------+----+
    // |[Columbus,Ohio]| Yin|
    // +---------------+----+
  }

  def runCsvDatasetExample(spark: SparkSession): Unit = {
    // csv_dataset
    // A CSV dataset is pointed to by path.
    // The path can be either a single CSV file or a directory of CSV files
    val path = "spark-sql-examples/src/main/resources/people.csv"

    val df = spark.read.csv(path)
    df.show()
    // +------------------+
    // |               _c0|
    // +------------------+
    // |      name;age;job|
    // |Jorge;30;Developer|
    // |  Bob;32;Developer|
    // +------------------+

    // Read a csv with delimiter, the default delimiter is ","
    val df2 = spark.read.option("delimiter", ";").csv(path)
    df2.show()
    // +-----+---+---------+
    // |  _c0|_c1|      _c2|
    // +-----+---+---------+
    // | name|age|      job|
    // |Jorge| 30|Developer|
    // |  Bob| 32|Developer|
    // +-----+---+---------+

    // Read a csv with delimiter and a header
    val df3 = spark.read.option("delimiter", ";").option("header", "true").csv(path)
    df3.show()
    // +-----+---+---------+
    // | name|age|      job|
    // +-----+---+---------+
    // |Jorge| 30|Developer|
    // |  Bob| 32|Developer|
    // +-----+---+---------+

    // You can also use options() to use multiple options
    val df4 = spark.read.options(Map("delimiter" -> ";", "header" -> "true")).csv(path)
    df4.show()
    // +-----+---+---------+
    // | name|age|      job|
    // +-----+---+---------+
    // |Jorge| 30|Developer|
    // |  Bob| 32|Developer|
    // +-----+---+---------+

    // "output" is a folder which contains multiple csv files and a _SUCCESS file.
    df4.write.option("header", "true").csv("tem/output")

    val df5 = spark.read.option("header", "true").csv("tem/output")
    df5.show()
    // +-----+---+---------+
    // | name|age|      job|
    // +-----+---+---------+
    // |Jorge| 30|Developer|
    // |  Bob| 32|Developer|
    // +-----+---+---------+
  }

  def runTextDatasetExample(spark: SparkSession): Unit = {
    // text_dataset
    // A text dataset is pointed to by path.
    // The path can be either a single text file or a directory of text files
    val path = "spark-sql-examples/src/main/resources/people.txt"

    val df1 = spark.read.text(path)
    df1.show()
    // +-----------+
    // |      value|
    // +-----------+
    // |Michael, 29|
    // |   Andy, 30|
    // | Justin, 19|
    // +-----------+

    // You can use 'lineSep' option to define the line separator.
    // The line separator handles all `\r`, `\r\n` and `\n` by default.
    val df2 = spark.read.option("lineSep", ",").text(path)
    df2.show()
    // +-----------+
    // |      value|
    // +-----------+
    // |    Michael|
    // |   29\nAndy|
    // | 30\nJustin|
    // |       19\n|
    // +-----------+

    // You can also use 'wholetext' option to read each input file as a single row.
    val df3 = spark.read.option("wholetext", true).text(path)
    df3.show()
    //  +--------------------+
    //  |               value|
    //  +--------------------+
    //  |Michael, 29\nAndy...|
    //  +--------------------+

    // "output" is a folder which contains multiple text files and a _SUCCESS file.
    df1.write.text("tem/output")

    // You can specify the compression format using the 'compression' option.
    df1.write.option("compression", "gzip").text("tem/output_compressed")
  }

  def runJdbcDatasetExample(spark: SparkSession): Unit = {
    // jdbc_dataset
    // Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
    // Loading data from a JDBC source
    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306")
      .option("dbtable", "school.student")
      .option("user", "root")
      .option("password", "bjfu1022")
      .load()
    jdbcDF.show()

    val connectionProperties = new Properties()
    connectionProperties.put("user", "root")
    connectionProperties.put("password", "bjfu1022")
    val jdbcDF2 = spark.read
      .jdbc("jdbc:mysql://localhost:3306", "school.student", connectionProperties)
    // Specifying the custom data types of the read schema
    jdbcDF2.show()

    // Saving data to a JDBC source
    jdbcDF.write
      .format("jdbc")
      .mode("overwrite")
      .option("url", "jdbc:mysql://localhost:3306")
      .option("dbtable", "school.student2")
      .option("user", "root")
      .option("password", "bjfu1022")
      .save()

    jdbcDF2.write
      .mode("overwrite")
      .jdbc("jdbc:mysql://localhost:3306", "school.student3", connectionProperties)


  }
}
